package com.atguigu.day4

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object RedirectLateEventCustom {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //    设置时间语义为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    env.setParallelism(1)
    //
    val stream1 = env.socketTextStream("192.168.35.102", 9999, '\n')
      .map(line => {
        var arr = line.split(" ")
        (arr(0), arr(1).toLong * 1000L)
      })
      .assignAscendingTimestamps(_._2)
      .process(new LastEventProc)

    stream1.print()
    stream1.getSideOutput(new OutputTag[String]("late")).print()
    env.execute()

  }

  class LastEventProc extends ProcessFunction[(String, Long), (String, Long)] {
    val late = new OutputTag[String]("late")

    override def processElement(value: (String, Long), ctx: ProcessFunction[(String, Long), (String, Long)]#Context, out: Collector[(String, Long)]): Unit = {

      //如果到来元素所包含的时间戳发送到侧输出流中
      if (value._2 < ctx.timerService().currentWatermark()) {
        ctx.output(late, s"迟到事件来了${value}")
      } else {
        out.collect(value)
      }
    }
  }
}
